public class LastHourTop5Cities {
    public static void main(String[] args) {
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "realtime_top5_city_app");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "****:9092,****:9093");

            Serde<String> stringSerde = Serdes.String();
            Serde<CityLevelCampaignReport> campaignSerde = Serdes.CampaignReportSerde();
            Serde<PriorityQueue> priorityQueueSerde = Serdes.PriorityQueueSerde();
            int topN = 5;

            StreamsBuilder builder = new StreamsBuilder();
            KStream<String, CityLevelCampaignReport> stream = builder.stream("campaign-last-hour", Consumed.with(stringSerde, campaignSerde));

            stream.filter(Objects::nonNull)
                .filter(campaign -> campaign.getCity() != null && campaign.getCampaign() != null)
                .groupBy((noKey, value) -> KeyValue.pair(CityLevelCampaignReport.getCampaign(), value), Grouped.with(stringSerde, campaignSerde))
                .windowedBy(TimeWindows.of(Duration.ofHours(1)))
                .aggregate(
                    () -> {
                        Comparator<CityLevelCampaignReport> comparator = (c1, c2) -> (int)(c1.getImpression() - c2.getImpression());
                        return new PriorityQueue<>(comparator);
                    },
                    (windowedCampaign, record, queue) -> {
                        queue.add(record);
                        return queue;
                    },
                    (windowedCampaign, record, queue) -> {
                        queue.remove(record);
                        return queue;
                    },
                    priorityQueueSerde,
                    "Top5Cities")
                .toStream()
                .mapValues(queue -> {
                    StringBuilder sb = new StringBuilder();
                    CityLevelCampaignReport record = queue.poll();
                    for (int i = 0; i < topN; i++) {
                        sb.append(record.getCity() + " -> " + record.getImpression());
                    }
                    return sb.toString();
                })
                .to("campaign-top5-city", Produced.with(stringSerde, stringSerde));

            KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), props);
            LOG.info("Start to calcuate Top 5 Cities for last hour");
            kafkaStreams.start();
            Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
    }
}